@CTTY do you have time to look at this?
Tried to make this change as simple as possible.
7 lines of functional code change, everything else is comments or test.

what the benefit of this is vs using the StaticTableProvider?

If we decide this is the right path forward, I wonder if we can make it generic such that this works for incremental append scans etc also? Maybe with a ScanRange enum or something? Happy to hear thoughts here!

My idea is to register tables to datafusion using IcebergTableProvider, so that the same registration also allows for inserts/deletes/merges, etc. Whereas StaticTableProvider is strictly scan only.
Also a staticTableProvider can never go back to current_state as it never refreshes, with my PR unpinning
with_snapshot_id(None)
will return back to current state.

There's also a distributed angle (my motivating use case is a Ballista integration): the flow is Client -> LogicalCodec serializes the provider's recipe -> Scheduler rebuilds the provider and does physical planning -> PhysicalCodec -> Executors. The provider needs to hold everything required to recreate the table connection on remote executors: snapshot_id here, and catalog config (catalog_type, name, storage_properties) in a follow-up PR.
This is split out from #2727, where the fuller picture is visible.

I think the ScanRange suggestion is great, and since the current snapshot_id field is private we can replace it with ScanRange in the future without breaking anything. This would also require some plumbing to IcebergTableScan to work, which I think should live in a separate PR to keep this minimal.
We can make a stub field too for the time being, and put snapshot_id in it?

Something seems a little off here to me. Read/write skew seems like a footgun. A single registered table that reads snapshot A but writes to HEAD is a confusing object. INSERT INTO t; SELECT * FROM t won't show the row you just inserted. That's a strange contract to hand a user in my opinion.

I also think schema isn't being handled here correctly for pinned snapshots, we should resolve against the snapshot's schema rather than current, the static table provider handles this correctly.

Something seems a little off here to me. Read/write skew seems like a footgun. A single registered table that reads snapshot A but writes to HEAD is a confusing object. INSERT INTO t; SELECT * FROM t won't show the row you just inserted. That's a strange contract to hand a user in my opinion.

I also think schema isn't being handled here correctly for pinned snapshots, we should resolve against the snapshot's schema rather than current, the static table provider handles this correctly.

insert-read problem:
Sorry for that, I understand what you are saying on careful review. I was so tunnel visioned on the ballista integration, that I had missed actual users directly calling snapshot_id.

The way I have it wired in my Ballista integration is that snapshot_id is resolved by the scheduler when it encodes the physical plan to executors, so a read -> insert -> read would properly show the new data as the snapshot_id would be reset to head by the scheduler as queries are sequentially made.

I think it makes sense to add a write guard, if the snapshot_id is pinned (not None), to constrain confusing behavior? I added it in a new commit, let me know how you think this should be handled.

Thanks for iterating with me! What are we gaining here over using the StaticTableProvider now?

Thanks for iterating with me! What are we gaining here over using the StaticTableProvider now?

Thanks for helping me with this too (still fixing the schema update with snapshot id problem), and sorry for responding late.

To answer your question, anyone not using the iceberg-ballista integration likely won't benefit from these changes, and they should keep using IcebergStaticTableProvider for time-travel reads. After this PR the two resolve schemas through the same helper and both refuse writes when pinned, so read semantics are identical.

Ballista requires you instantiate one TableProvider, through which queries are routed (explained above). This creates two requirements:

1. The TableProvider needs to carry all the information required to rebuild it on the scheduler, which de-serializes the logical plan and re-plans it (this PR + another one add the missing requisite information: snapshot-id + catalog-configuration). StaticTableProvider is built from an already-loaded Table and carries no catalog identity, so there is nothing a codec can rebuild it from.
2. If I want the same instantiated TableProvider to allow for both Scanning + Insert + etc, it needs functional .scan() and .insert_into() implementations. StaticTableProvider has the methods (the trait requires them), but its .insert_into() unconditionally errors, so writes are structurally impossible for it.

Schema Problem:
Following StaticTableProvider's approach (per your suggestion), pinning now re-derives the exposed schema from the pinned snapshot.

**Also now re-loads the table to reload the schema.

While addressing this, I found a second issue: with_snapshot_id cached the Arrow schema at construction time (the current schema). Pinning across a schema evolution would then push down non-existent column names from the old schema.

Fix: Schema resolution is now centralized in a shared snapshot_arrow_schema helper that looks up the snapshot’s own schema via snapshot.schema(metadata). Both with_snapshot_id and IcebergStaticTableProvider constructors use this helper, ensuring a single resolution path. with_snapshot_id is now fallible, as the snapshot ID is validated during pinning.

Hey @xanderbailey can you take a look at this again when you have time?
Hopefully it is in a more acceptable state.

Run with me here for a moment if you will. Does something like this not work for ballista?

  pub struct IcebergSerializableTableProvider {
      table_ident: TableIdent,
      metadata: TableMetadata,
      storage_props: HashMap<String, String>,
      storage_factory: Arc<dyn StorageFactory>,
      snapshot_id: Option<i64>,
      inner: IcebergStaticTableProvider,
  }

If we were to add something like this as the table provider, everything here apart from inner is serialisable which means you can fully construct it on the "executor side" (sorry not sure what ballista calls these). This keeps the current public API clear of any runtime errors that may happen as a result of people trying to pin for scans and fail on read. My understanding is that the IcebergStaticTableProvider is designed for exactly the use-case you're describing.

This keeps the current public API clear of any runtime errors that may happen as a result of people trying to pin for scans and fail on read

( Did you mean pin for scans and fail on inserts?)

Thanks for the response. Let me give a clearer description of what Ballista is doing
here, since the execution model isn't obvious.

To register a table with a Ballista context, the client provides a TableProvider:

pub async fn register_iceberg_table(
    ctx: &SessionContext,
    register_name: &str,
    config: IcebergCatalogConfig,
    namespace: NamespaceIdent,
    table: impl Into<String>,
) -> Result<(), DataFusionError> {
    let catalog = bridge::build_catalog(&config).await?;
    let provider = iceberg_datafusion::IcebergTableProvider::try_new_with_config(
        catalog, config, namespace, table,
    )
    .await
    .map_err(bridge::to_df_err)?;
    ctx.register_table(register_name, Arc::new(provider))?;
    Ok(())
}

From there:

Client     -> LogicalCodec::try_encode     (extract serializable fields from the provider)
Scheduler  -> LogicalCodec::try_decode     (rebuild the provider)
Scheduler  -> create_physical_plan()
                -> provider.scan() / .insert_into()
                -> IcebergTableScan  /  (IcebergWriteExec -> IcebergCommitExec)
Scheduler  -> physical optimizer rules
Scheduler  -> split into stages at shuffle boundaries
Scheduler  -> per stage: PhysicalCodec::try_encode -> ship to executors
Executor   -> decode, run one partition

The provider never reaches an executor. It's rebuilt on the scheduler, used for a single
create_physical_plan() call, and stays there. Executors only ever receive encoded
ExecutionPlan nodes.

At try_decode the codec can't tell a read from a write. datafusion-proto encodes DML
write targets as synthetic TableScan nodes, so scans and inserts arrive in an identical
shape. That rules out having the codec build an IcebergStaticTableProvider for reads and
an IcebergTableProvider for writes, since the information isn't there.

On the struct as sketched: storage_factory is Arc<dyn StorageFactory>, a trait object,
so it can't be serialized. And StorageConfig is a props map with no scheme field, so
storage_props alone doesn't say which backend to rebuild. So "everything apart from
inner is serialisable" unfortunately doesn't hold as written.

That said, I've been investigating your suggestion in this shape:

pub struct IcebergSerializableTableProvider {
    snapshot_id: Option<i64>,
    catalog_config: IcebergCatalogConfig,   // from the other PR
    schema: SchemaRef,
    inner: IcebergTableProvider,
}

Most methods delegate to inner. scan() and schema() are overridden to apply the
pinned snapshot. This keeps the distributed concerns in the Ballista crate, which I think
is what you're after.

It will still require some of the upstream change listed in NoahKusaba:feature/datafusion-catalog-config-v2 such as:

// table/mod.rs, to construct `inner`
pub(crate) async fn try_new(..)                -> pub

// physical_plan/scan.rs, to rebuild a scan on the executor
pub(crate) fn new(..)                          -> pub
pub fn with_predicates(..)                     // new. `new` derives the predicate from
                                               // Expr filters, but on decode we already
                                               // hold a Predicate and can't turn it back

// physical_plan/mod.rs, these are pub(crate) structs and so currently unnameable
pub use commit::IcebergCommitExec;
pub use write::IcebergWriteExec;
pub use metadata_scan::IcebergMetadataScan;
pub use project::PartitionExpr;

I still need to more deeply investigate the full lift here.

So the tradeoff is that the wrapper keeps new fields out of iceberg-datafusion, at the cost
of an extra abstraction layer + making execution-node constructors API public (which I needed to do anyway).

I'd also note the pieces in these PRs seem useful beyond Ballista.
IcebergStaticTableProvider supports time travel but has no catalog backing, so there's
currently no way to pin a snapshot on a catalog-registered table. My original instinct was
to put this in iceberg-datafusion so other engines could reuse it and to avoid the extra
abstraction layer, but I understand if you'd rather keep distributed concerns out of the
crate.

On the runtime-error concern specifically: an alternative is to have insert_into reset
snapshot_id to None and reload the schema, rather than erroring. That trades the error
for behavior some users might find surprising, so I don't have a strong preference. Happy
to go whichever way you prefer.

If we don't want to merge the snapshot change, I will persue the thread of keeping more distributed required features in the ballista-iceberg crate instead.